home *** CD-ROM | disk | FTP | other *** search
/ Personal Computer World 2009 February / PCWFEB09.iso / Software / Resources / Chat & Communication / Digsby build 37 / digsby_setup.exe / lib / jabber / threadstream.pyo (.txt) < prev    next >
Python Compiled Bytecode  |  2008-10-13  |  19KB  |  653 lines
  1. # Source Generated with Decompyle++
  2. # File: in.pyo (Python 2.5)
  3.  
  4. from __future__ import with_statement
  5. import pyxmpp.error as pyxmpp
  6. from dns import resolver
  7. from pyxmpp.exceptions import ClientStreamError
  8. from util.threads.threadpool2 import threaded
  9. from util.callbacks import callsback
  10. from util import GetSocketType, Delegate
  11. from pyxmpp.exceptions import LegacyAuthenticationError
  12. from pyxmpp.streambase import STREAM_NS
  13. from util.diagnostic import Diagnostic
  14. import libxml2
  15. import time
  16. import Queue
  17. import logging
  18. import socket
  19. from jabber.threadstreamsocket import ThreadStreamSocket
  20. from pyxmpp.jabber.clientstream import LegacyClientStream
  21. from pyxmpp.exceptions import StreamError, StreamEncryptionRequired, HostMismatch, ProtocolError, TLSError
  22. from pyxmpp.exceptions import FatalStreamError, StreamParseError, StreamAuthenticationError, SASLAuthenticationFailed
  23. from pyxmpp.jid import JID
  24. from pyxmpp import resolver
  25. from common import netcall
  26. from threading import currentThread
  27. import traceback
  28. import sys
  29. log = logging.getLogger('ThreadStream')
  30. outdebug = logging.getLogger('ThreadStream.out').debug
  31. outdebug_s = getattr(logging.getLogger('ThreadStream.out'), 'debug_s', outdebug)
  32. indebug = logging.getLogger('ThreadStream.in').debug
  33. indebug_s = getattr(logging.getLogger('ThreadStream.in'), 'debug_s', outdebug)
  34.  
  35. try:
  36.     import M2Crypto
  37.     if M2Crypto.version_info < (0, 16):
  38.         tls_available = 0
  39.     else:
  40.         from M2Crypto import SSL
  41.         from M2Crypto.SSL import SSLError
  42.         import M2Crypto.SSL.cb as M2Crypto
  43.         tls_available = 1
  44. except ImportError:
  45.     tls_available = 0
  46.  
  47.  
  48. class ThreadStream(LegacyClientStream):
  49.     
  50.     def __init__(self, *a, **k):
  51.         LegacyClientStream.__init__(self, *a, **k)
  52.         self._ThreadStream__logger = logging.getLogger('ThreadStream')
  53.         self._ThreadStream__shutdown = False
  54.         self.on_incoming_node = Delegate()
  55.         self.on_outgoing_node = Delegate()
  56.  
  57.     
  58.     def stanza(self, _unused, node):
  59.         self.on_incoming_node(node)
  60.         LegacyClientStream.stanza(self, _unused, node)
  61.  
  62.     
  63.     def _write_node(self, xmlnode):
  64.         self.on_outgoing_node(xmlnode)
  65.         LegacyClientStream._write_node(self, xmlnode)
  66.  
  67.     
  68.     def write_raw(self, data):
  69.         (None, netcall)((lambda : LegacyClientStream.write_raw(self, data)))
  70.  
  71.     
  72.     def idle(self):
  73.         (netcall,)((lambda : LegacyClientStream.idle(self)))
  74.  
  75.     
  76.     def send(self, stanza):
  77.         (None, netcall)((lambda : LegacyClientStream.send(self, stanza)))
  78.  
  79.     
  80.     def _write_raw(self, data):
  81.         if sys.DEV and currentThread().getName() != 'AsyncoreThread':
  82.             
  83.             try:
  84.                 raise AssertionError, 'bad thread for _write_raw: %r' % currentThread().getName()
  85.             except AssertionError:
  86.                 traceback.print_exc()
  87.                 traceback.print_stack()
  88.                 import wx
  89.                 
  90.                 def do_submit():
  91.                     d = Diagnostic(description = 'Automated: Woah, bad thread')
  92.                     d.prepare_data()
  93.                     d.do_post()
  94.                     profile = profile
  95.                     import common
  96.                     uname = profile.username
  97.                     del profile
  98.                     wx.MessageBox('Hey %s! Something crazy just happened!\nI submitted a bug report for you. - Chris' % uname)
  99.  
  100.                 wx.CallLater(3000, do_submit)
  101.                 raise 
  102.             except:
  103.                 None<EXCEPTION MATCH>AssertionError
  104.             
  105.  
  106.         None<EXCEPTION MATCH>AssertionError
  107.         outdebug_s('OUT: %r', data)
  108.         
  109.         try:
  110.             self.socket.push(data)
  111.         except Exception:
  112.             e = None
  113.             self.handle_error(e)
  114.  
  115.         outdebug('OUT: done')
  116.  
  117.     
  118.     def fileno(self):
  119.         self.lock.__enter__()
  120.         
  121.         try:
  122.             return self.socket._fileno
  123.         finally:
  124.             pass
  125.  
  126.  
  127.     
  128.     def connect(self, server = None, port = None):
  129.         outdebug('connect')
  130.         self.lock.__enter__()
  131.         
  132.         try:
  133.             self._connect1(server, port)
  134.         finally:
  135.             pass
  136.  
  137.  
  138.     
  139.     def _connect1(self, server = None, port = None):
  140.         outdebug('_connect1')
  141.         if not (self.my_jid.node) or not (self.my_jid.resource):
  142.             raise ClientStreamError, 'Client JID must have username and resource'
  143.         
  144.         if not server:
  145.             server = self.server
  146.         
  147.         if not port:
  148.             port = self.port
  149.         
  150.         if server:
  151.             self._ThreadStream__logger.debug('server: %r', (server, port))
  152.             service = None
  153.         else:
  154.             service = 'xmpp-client'
  155.         if port is None:
  156.             port = 5222
  157.         
  158.         if server is None:
  159.             self._ThreadStream__logger.debug('server: %r', (server, port))
  160.             server = self.my_jid.domain
  161.         
  162.         self.me = self.my_jid
  163.         
  164.         def connect_failed():
  165.             self.owner.set_offline(self.owner.Reasons.CONN_FAIL)
  166.  
  167.         self._connect2(server, port, service, self.my_jid.domain, sck_cls = GetSocketType())
  168.  
  169.     
  170.     def _connect2(self, addr1, port1, service = None, to = None, sck_cls = socket.SocketType):
  171.         outdebug('_connect2')
  172.         self._ThreadStream__logger.debug('server: %r', (addr1, port1))
  173.         if to is None:
  174.             to = str(addr1)
  175.         
  176.         if service is not None:
  177.             self.state_change('resolving srv', (addr1, service))
  178.             
  179.             try:
  180.                 addrs = resolver.resolve_srv(addr1, service)
  181.             except Exception:
  182.                 traceback.print_exc()
  183.                 addrs = []
  184.  
  185.             if not addrs:
  186.                 addrs = [
  187.                     (addr1, port1)]
  188.             else:
  189.                 addrs.append((addr1, port1))
  190.         else:
  191.             addrs = [
  192.                 (addr1, port1)]
  193.         msg = None
  194.         self._ThreadStream__logger.debug('addrs: %r', addrs)
  195.         for addr, port in addrs:
  196.             if type(addr) in (str, unicode):
  197.                 self.state_change('resolving', addr)
  198.             
  199.             s = None
  200.             
  201.             try:
  202.                 resolved = resolver.getaddrinfo(addr, port, 0, socket.SOCK_STREAM)
  203.             except Exception:
  204.                 traceback.print_exc()
  205.                 resolved = []
  206.  
  207.             resolved.append((2, 1, 0, '_unused', (addr, port)))
  208.             for res in resolved:
  209.                 (family, socktype, proto, _unused, sockaddr) = res
  210.                 self._ThreadStream__logger.debug('res: %r', res)
  211.                 
  212.                 try:
  213.                     s = sck_cls(family, socktype, proto)
  214.                     s.settimeout(10)
  215.                     self.state_change('connecting', sockaddr)
  216.                     s.connect(sockaddr)
  217.                     if self.owner.do_ssl:
  218.                         ctx = SSL.Context()
  219.                         ctx.set_verify(SSL.verify_none, 10)
  220.                         s.setblocking(True)
  221.                         ssl = SSL.Connection(ctx, s)
  222.                         ssl.setup_ssl()
  223.                         ssl.set_connect_state()
  224.                         ssl.connect_ssl()
  225.                         s.setblocking(False)
  226.                         s = ssl
  227.                         s.setblocking(False)
  228.                     
  229.                     self.state_change('connected', sockaddr)
  230.                 except (socket.error, SSLError):
  231.                     msg = None
  232.                     self._ThreadStream__logger.debug('Connect to %r failed: %r', sockaddr, msg)
  233.                     traceback.print_exc()
  234.                     if s:
  235.                         s.close()
  236.                         s = None
  237.                         continue
  238.                     continue
  239.  
  240.                 break
  241.             
  242.             if s:
  243.                 self._ThreadStream__logger.debug('connected to: %r', (addr, port))
  244.                 break
  245.                 continue
  246.         
  247.         if not s:
  248.             if msg:
  249.                 self._ThreadStream__logger.debug('failed to connect to %r: %r', (addr, port), msg)
  250.                 raise socket.error, msg
  251.             else:
  252.                 self._ThreadStream__logger.debug('failed to connect to %r: unknown reason', (addr, port))
  253.                 raise FatalStreamError, 'Cannot connect'
  254.         
  255.         self.addr = addr
  256.         self.port = port
  257.         self.owner.lock.__enter__()
  258.         
  259.         try:
  260.             if self.owner.connect_killed == True:
  261.                 raise FatalStreamError, 'Connect Killed'
  262.         finally:
  263.             pass
  264.  
  265.         self._connect_socket(s, to)
  266.         self.last_keepalive = time.time()
  267.  
  268.     
  269.     def closed(self):
  270.         self.owner.fatal_error()
  271.         self.state_change('disconnected', self.peer)
  272.  
  273.     
  274.     def closed_dead(self):
  275.         self.owner.fatal_error()
  276.         self.close(False)
  277.         self.owner.disconnected()
  278.  
  279.     
  280.     def __connect_error(self):
  281.         pass
  282.  
  283.     
  284.     def _connect_socket(self, sock, to = None):
  285.         logging.getLogger('ThreadStream').debug('connecting')
  286.         new_sock = ThreadStreamSocket(sock, self._feed_reader, 100, self.closed, self.closed_dead, ssl = self.owner.do_ssl)
  287.         (None, None, netcall)((lambda : LegacyClientStream._connect_socket(self, new_sock, to)))
  288.  
  289.     
  290.     def _loop_iter(self, timeout):
  291.         pass
  292.  
  293.     
  294.     def _process(self):
  295.         pass
  296.  
  297.     
  298.     def _read(self):
  299.         pass
  300.  
  301.     
  302.     def _process_tls_node(self, xmlnode):
  303.         if not (self.tls_settings) or not tls_available:
  304.             self._ThreadStream__logger.debug('Unexpected TLS node: %r' % xmlnode.serialize())
  305.             return False
  306.         
  307.         if self.initiator:
  308.             if xmlnode.name == 'failure':
  309.                 raise TLSNegotiationFailed, 'Peer failed to initialize TLS connection'
  310.             elif xmlnode.name != 'proceed' or not (self.tls_requested):
  311.                 self._ThreadStream__logger.debug('Unexpected TLS node: %r' % xmlnode.serialize())
  312.                 return False
  313.             
  314.             self.tls_requested = 0
  315.             self._make_tls_connection(success = self.finish_process, error = self.fail_process)
  316.         
  317.         return True
  318.  
  319.     
  320.     def fail_process(self):
  321.         self.owner.fatal_error()
  322.         self.close()
  323.  
  324.     
  325.     def finish_process(self):
  326.         self.socket = self.tls
  327.         self._ThreadStream__logger.debug('Restarting XMPP stream')
  328.         self._restart_stream()
  329.         return True
  330.  
  331.     
  332.     def _make_tls_connection(self, callback = None):
  333.         ctx = None
  334.         
  335.         try:
  336.             if not tls_available or not (self.tls_settings):
  337.                 raise TLSError, 'TLS is not available'
  338.             
  339.             tlssettings = self.tls_settings
  340.             self.state_change('tls connecting', self.peer)
  341.             self._ThreadStream__logger.debug('Creating TLS context')
  342.             ctx = None if tlssettings.ctx else SSL.Context('tlsv1')
  343.             verify_callback = tlssettings.verify_callback
  344.             if not verify_callback:
  345.                 verify_callback = self.tls_default_verify_callback
  346.             
  347.             if tlssettings.verify_peer:
  348.                 self._ThreadStream__logger.debug('verify_peer, verify_callback: %r', verify_callback)
  349.                 ctx.set_verify(SSL.verify_peer, 10, verify_callback)
  350.             else:
  351.                 ctx.set_verify(SSL.verify_none, 10)
  352.             if tlssettings.cert_file:
  353.                 ctx.use_certificate_chain_file(tlssettings.cert_file)
  354.                 if tlssettings.key_file:
  355.                     ctx.use_PrivateKey_file(tlssettings.key_file)
  356.                 else:
  357.                     ctx.use_PrivateKey_file(tlssettings.cert_file)
  358.                 ctx.check_private_key()
  359.             
  360.             if tlssettings.cacert_file:
  361.                 
  362.                 try:
  363.                     ctx.load_verify_location(tlssettings.cacert_file)
  364.                 except AttributeError:
  365.                     ctx.load_verify_locations(tlssettings.cacert_file)
  366.                 except:
  367.                     None<EXCEPTION MATCH>AttributeError
  368.                 
  369.  
  370.             None<EXCEPTION MATCH>AttributeError
  371.         except:
  372.             callback.error()
  373.             return None
  374.  
  375.         self.callback = callback
  376.         self.socket.make_tls(ctx, success = self.tls_done, error = self.tls_fail)
  377.  
  378.     _make_tls_connection = callsback(_make_tls_connection)
  379.     tls_fail = fail_process
  380.     
  381.     def tls_done(self):
  382.         self.tls = self.socket
  383.         self.state_change('tls connected', self.peer)
  384.         
  385.         try:
  386.             raise Exception
  387.         except:
  388.             pass
  389.  
  390.         self.callback.success()
  391.  
  392.     
  393.     def _got_features(self):
  394.         
  395.         try:
  396.             return LegacyClientStream._got_features(self)
  397.         except FatalStreamError:
  398.             e = None
  399.             if e.__class__ == FatalStreamError:
  400.                 self.owner.auth_failed(e.message)
  401.             else:
  402.                 raise 
  403.         except:
  404.             e.__class__ == FatalStreamError
  405.  
  406.  
  407.     
  408.     def registration_error(self, stanza):
  409.         self.lock.__enter__()
  410.         
  411.         try:
  412.             ae = None
  413.             err = stanza.get_error()
  414.             ae = err.xpath_eval('e:*', {
  415.                 'e': 'jabber:iq:auth:error' })
  416.             if ae:
  417.                 ae = ae[0].name
  418.             else:
  419.                 ae = err.get_condition().name
  420.         finally:
  421.             pass
  422.  
  423.         if self.registration_error_callback is not None:
  424.             self.registration_error_callback((ae,) + pyxmpp.error.stanza_errors[ae])
  425.         
  426.         self.registration_error_callback = None
  427.         self.registration_success_callback = None
  428.  
  429.     
  430.     def registration_success(self, stanza):
  431.         if self.registration_success_callback is not None:
  432.             self.registration_success_callback()
  433.         
  434.         self.registration_success_callback = None
  435.         self.registration_error_callback = None
  436.         _unused = stanza
  437.         self.lock.__enter__()
  438.         
  439.         try:
  440.             self.state_change('registered', self.registration_form)
  441.             if 'FORM_TYPE' in self.registration_form and self.registration_form['FORM_TYPE'].value == 'jabber:iq:register':
  442.                 if 'username' in self.registration_form:
  443.                     self.my_jid = JID(self.registration_form['username'].value, self.my_jid.domain, self.my_jid.resource)
  444.                 
  445.                 if 'password' in self.registration_form:
  446.                     self.password = self.registration_form['password']
  447.                 
  448.             
  449.             self.registration_callback = None
  450.         finally:
  451.             pass
  452.  
  453.  
  454.     
  455.     def disconnect(self):
  456.         LegacyClientStream.disconnect(self)
  457.         self.state_change('disconnected', self.peer)
  458.  
  459.     
  460.     def stream_end(self, _unused):
  461.         LegacyClientStream.stream_end(self, _unused)
  462.         self.shutdown()
  463.  
  464.     
  465.     def _send_stream_end(self):
  466.         LegacyClientStream._send_stream_end(self)
  467.         self.shutdown()
  468.  
  469.     
  470.     def shutdown(self):
  471.         if not self._ThreadStream__shutdown:
  472.             outdebug('non-Force shutdown')
  473.             self._ThreadStream__shutdown = True
  474.             if self.socket:
  475.                 outdebug('non-Force close_when_done')
  476.                 self.socket.close_when_done()
  477.             
  478.         else:
  479.             outdebug('Force shutdown')
  480.             self.close(False)
  481.  
  482.     
  483.     def close(self, do_disconnect = True):
  484.         self.lock.__enter__()
  485.         
  486.         try:
  487.             return self._close(do_disconnect)
  488.         finally:
  489.             pass
  490.  
  491.  
  492.     
  493.     def _close(self, do_disconnect = True):
  494.         if do_disconnect:
  495.             self._disconnect()
  496.         
  497.         if self.doc_in:
  498.             self.doc_in = None
  499.         
  500.         if self.features:
  501.             self.features = None
  502.         
  503.         self._reader = None
  504.         self.stream_id = None
  505.         if self.socket:
  506.             self.socket.close()
  507.         
  508.         self._reset()
  509.  
  510.     
  511.     def _process_node(self, stanza):
  512.         
  513.         try:
  514.             LegacyClientStream._process_node(self, stanza)
  515.         except SASLAuthenticationFailed:
  516.             e = None
  517.             self.owner.auth_failed(reason = e.message)
  518.             self._ThreadStream__logger.critical('SASLAuthenticationFailed')
  519.         except LegacyAuthenticationError:
  520.             e = None
  521.             self.owner.auth_failed(reason = e.message)
  522.             self._ThreadStream__logger.critical('LegacyAuthenticationError')
  523.         except FatalStreamError:
  524.             e = None
  525.             import hub as hub
  526.             hub.get_instance().on_error(e)
  527.             self._ThreadStream__logger.critical('Stream blew up')
  528.             self.owner.fatal_error()
  529.             self.close()
  530.  
  531.  
  532.     
  533.     def error(self, descr):
  534.         self._ThreadStream__logger.critical('XML parse error: ' + descr)
  535.         self.owner.fatal_error()
  536.         self.close()
  537.  
  538.     
  539.     def fix_in_stanza(self, stanza):
  540.         LegacyClientStream.fix_in_stanza(self, stanza)
  541.         if self.initiator:
  542.             to = stanza.get_to()
  543.             if to is not None:
  544.                 p = self.peer
  545.                 pb = None if p else None
  546.                 tob = None if to else None
  547.                 if tob == pb and to == p and to == pb or tob == p:
  548.                     stanza.set_to(False)
  549.                 
  550.             
  551.         
  552.  
  553.     
  554.     def _feed_reader(self, data):
  555.         self.lock.__enter__()
  556.         
  557.         try:
  558.             if self._reader is not None:
  559.                 self._super_feed_reader(data)
  560.             else:
  561.                 self.close(False)
  562.         finally:
  563.             pass
  564.  
  565.  
  566.     
  567.     def _super_feed_reader(self, data):
  568.         indebug_s('IN: %r', data)
  569.         if data:
  570.             
  571.             try:
  572.                 r = self._reader.feed(data)
  573.                 while r:
  574.                     r = self._reader.feed('')
  575.                 if r is None:
  576.                     indebug('r was None, setting eof + disconnect')
  577.                     self.eof = 1
  578.                     self.disconnect()
  579.             except StreamParseError:
  580.                 self._send_stream_error('xml-not-well-formed')
  581.                 raise 
  582.             except:
  583.                 None<EXCEPTION MATCH>StreamParseError
  584.             
  585.  
  586.         None<EXCEPTION MATCH>StreamParseError
  587.         indebug('no data, setting eof + disconnect')
  588.         self.eof = 1
  589.         self.disconnect()
  590.         if self.eof:
  591.             indebug('eof calling stream_end')
  592.             self.stream_end(None)
  593.         
  594.  
  595.     
  596.     def stream_start(self, doc):
  597.         self.doc_in = doc
  598.         log.debug('input document: %r' % (self.doc_in.serialize(),))
  599.         
  600.         try:
  601.             r = self.doc_in.getRootElement()
  602.             if r.ns().getContent() != STREAM_NS:
  603.                 self._send_stream_error('invalid-namespace')
  604.                 raise FatalStreamError, 'Invalid namespace.'
  605.         except libxml2.treeError:
  606.             self._send_stream_error('invalid-namespace')
  607.             raise FatalStreamError, "Couldn't get the namespace."
  608.  
  609.         self.version = r.prop('version')
  610.         if self.version and self.version != '1.0':
  611.             self._send_stream_error('unsupported-version')
  612.             raise FatalStreamError, 'Unsupported protocol version.'
  613.         
  614.         to_from_mismatch = 0
  615.         if self.initiator:
  616.             self.stream_id = r.prop('id')
  617.             peer = r.prop('from')
  618.             if peer:
  619.                 peer = JID(peer)
  620.             
  621.             if self.peer:
  622.                 if peer and peer != self.peer and not unicode(self.peer).endswith(unicode(peer)):
  623.                     self._ThreadStream__logger.debug('peer hostname mismatch: %r != %r' % (peer, self.peer))
  624.                     to_from_mismatch = 1
  625.                 elif peer:
  626.                     self.peer = peer
  627.                 
  628.             else:
  629.                 self.peer = peer
  630.         else:
  631.             to = r.prop('to')
  632.             if to:
  633.                 to = self.check_to(to)
  634.                 if not to:
  635.                     self._send_stream_error('host-unknown')
  636.                     raise FatalStreamError, 'Bad "to"'
  637.                 
  638.                 self.me = JID(to)
  639.             
  640.             self._send_stream_start(self.generate_id())
  641.             self._send_stream_features()
  642.             self.state_change('fully connected', self.peer)
  643.             self._post_connect()
  644.         if not self.version:
  645.             self.state_change('fully connected', self.peer)
  646.             self._post_connect()
  647.         
  648.         if to_from_mismatch:
  649.             raise HostMismatch
  650.         
  651.  
  652.  
  653.